package com.gitee.jmash.websocket.endpoint;

import java.io.IOException;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.gitee.jmash.core.utils.JsonbUtil;
import com.gitee.jmash.mvc.CdiAwareConfigurator;
import com.gitee.jmash.websocket.model.Message;
import com.gitee.jmash.websocket.model.MessageDecoder;
import com.gitee.jmash.websocket.model.MessageEncoder;
import jakarta.enterprise.context.Dependent;
import jakarta.websocket.EncodeException;
import jakarta.websocket.OnClose;
import jakarta.websocket.OnError;
import jakarta.websocket.OnMessage;
import jakarta.websocket.OnOpen;
import jakarta.websocket.Session;
import jakarta.websocket.server.ServerEndpoint;

/**
 * Websocket 订阅服务
 * 
 * @author cgd
 */
@ServerEndpoint(value = "/websocket", configurator = CdiAwareConfigurator.class,
    encoders = {MessageEncoder.class}, decoders = {MessageDecoder.class})
@Dependent
public class SubscribeEndpoint {

  private static Log log = LogFactory.getLog(SubscribeEndpoint.class);

  private static Map<String, Session> sessions = new ConcurrentHashMap<>();

  /** Server OnOpen. */
  @OnOpen
  public void onOpen(Session session) throws IOException, EncodeException {
    log.warn("Session 创建 ID：" + session.getId());
    sessions.put(session.getId(), session);
  }

  /** Server OnMessage. */
  @OnMessage
  public void onMessage(Session session, Message message) throws IOException, EncodeException {
    log.info("req:"+JsonbUtil.toJson(message));
    log.info("Session " + session.getId() + "  " + message.getMessage());
    SubscribeHandle.handle(session, message);
  }

  /** Server Close. */
  @OnClose
  public void onClose(Session session) {
    log.warn("Session 销毁 ID：" + session.getId());
    sessions.remove(session.getId());
    SubscribeHandle.close(session);
  }

  @OnError
  public void onError(Session session, Throwable tshrowable) {
    log.error("There has been an error with session " + session.getId());
  }

  /**
   * 广播消息(异步发送).
   * 
   * @param message 消息内容.
   */
  public static synchronized int broadcast(Message message) {
    int i = 0;
    for (Session session : sessions.values()) {
      session.getAsyncRemote().sendObject(message);
      i += 1;
    }
    return i;
  }

  /**
   * 发送消息.
   *
   * @param message 消息内容.
   */
  public static synchronized int sendMessages(Set<String> sessionIds, Message message,
      boolean sync) {
    int i = 0;
    for (String seesionId : sessionIds) {
      boolean success = sendMessage(seesionId, message, sync);
      if (success) {
        i++;
      }
    }
    return i;
  }

  /**
   * 发送消息.
   *
   * @param message 消息内容.
   */
  public static synchronized boolean sendMessage(String sessionId, Message message, boolean sync) {
    try {
      if (sessions.containsKey(sessionId)) {
        Session session = sessions.get(sessionId);
        if (sync) {
          session.getBasicRemote().sendObject(message);
        } else {
          session.getAsyncRemote().sendObject(message);
        }
        return true;
      }
    } catch (IOException | EncodeException e) {
      log.error("send message error:", e);
    }
    return false;
  }

}
